package dev_ops.tools.kafka;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Random;
import java.util.TreeMap;

import org.dev_common.func.Functions;
import org.dev_common.json.Jacksons;


public class ReplicaResign {
    public static void main(String[] args) throws Exception {
        Map<String, Map<String, List<Integer>>> resigns = getresigns();
        for (Entry<String, Map<String, List<Integer>>> entry : resigns.entrySet()) {
            String topic = entry.getKey();
            Map<String, List<Integer>> pair = entry.getValue();
            R r = new R();
            List<A> as = new ArrayList<>();
            for (Entry<String, List<Integer>> parts : pair.entrySet()) {
                A a = new A();
                a.setTopic(topic);
                a.setPartition(Integer.valueOf(parts.getKey().trim()));
                a.setReplicas(parts.getValue());
                as.add(a);
            }
            r.setPartitions(as);
            System.out.println(Jacksons.string(r));
            Files.write(Paths.get("/Users/thomas/kafka/"+topic), Jacksons.string(r).getBytes());
        }
    }

    static class R {
        private String version = "1.0";
        private List<A> partitions;

        public List<A> getPartitions() {
            return partitions;
        }

        public void setPartitions(List<A> partitions) {
            this.partitions = partitions;
        }

        public String getVersion() {
            return version;
        }

        public void setVersion(String version) {
            this.version = version;
        }
    }

    static class A {
        private String topic;
        private Integer partition;
        private List<Integer> replicas;

        public List<Integer> getReplicas() {
            return replicas;
        }

        public void setReplicas(List<Integer> replicas) {
            this.replicas = replicas;
        }

        public Integer getPartition() {
            return partition;
        }

        public void setPartition(Integer partition) {
            this.partition = partition;
        }

        public String getTopic() {
            return topic;
        }

        public void setTopic(String topic) {
            this.topic = topic;
        }
    }

    /**
     * 解析kafka topic --describe输出
     * @return
     * @throws IOException
     */
    private static Map<String, Map<String, List<Integer>>> getresigns() throws IOException {
        //List<String> lines = Files.readAllLines(Paths.get("/Users/thomas/kafka.conf"));
        List<String> lines = Files.readAllLines(Paths.get("/Users/thomas/topic_2"));
        Map<String, Map<String, List<Integer>>> cont = new HashMap<>();
        for (String line : lines) {
            String topic = "noexsit";
            if (!Functions.isBlank(line) && !line.startsWith("Topic")) {
                String[] splits = line.split("\t");
                topic = splits[1].replace("Topic:", "").trim();
                if (cont.get(topic) == null) {
                    Map<String, List<Integer>> map = new TreeMap<>();
                    cont.put(topic, map);
                }
                String part = splits[2].replace("Partition:", "");
                String leader = splits[3].replace("Leader:", "");
                List<Integer> replicas = new ArrayList<>();
                System.out.println(String.format("T:%s, P:%s, R:%s", topic, part, leader));
                if ("-1".equals(leader.trim())) {
                    replicas.add(4);
                }else {
                    replicas.add(Integer.valueOf(leader.trim()));
                }
                String rep = addMore(leader.trim());
                replicas.add(Integer.valueOf(rep));
                cont.get(topic).put(part, replicas);
            }
        }
        return cont;
    }
    
    /**
     * partition 分配策略
     */
    private static String addMore(String part) {
        int partition = Integer.valueOf(part);
        String ret = "-1";
        Random random = new Random();
        switch (partition) {
        case 1:
            ret = new String[] { "3", "4" }[random.nextInt(2)];
            break;
        case 2:
            ret = new String[] { "3", "4", "1" }[random.nextInt(3)];
            break;
        case 3:
            ret = new String[] { "1", "4" }[random.nextInt(2)];
            break;
        case 4:
            ret = new String[] { "3", "1" }[random.nextInt(2)];
            break;
        case -1:
            ret = new String[] { "3", "1" }[random.nextInt(2)];
            break;
        default:
            break;
        }
        return ret;
    }
}
